package com.tuanzhang.zhw.rpc.client;

import com.tuanzhang.zhw.rpc.codec.RpcRequest;
import com.tuanzhang.zhw.rpc.codec.RpcResponse;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.locks.AbstractQueuedSynchronizer;
import java.util.concurrent.locks.ReentrantLock;

@Slf4j
public class RpcFuture implements Future<Object> {

    private final static long TIME_THRESHOLD = 5000;

    private RpcRequest request;
    private RpcResponse response;
    private Sync sync;
    private long startTime;

    private List<RpcCallback> pendingCallbacks = new ArrayList<>();
    private ReentrantLock callbackLock = new ReentrantLock();
    private ThreadPoolExecutor executor = new ThreadPoolExecutor(16, 16, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(65536));

    public RpcFuture(RpcRequest request) {
        this.request = request;
        this.startTime = System.currentTimeMillis();
        sync = new Sync();
    }

    /**
     * 异步阻塞 回调处理
     * @param rpcResponse
     */
    public void done(RpcResponse rpcResponse) {
        this.response = rpcResponse;
        boolean success = sync.release(1);
        if (success) {
            invokeCallbacks();
        }
        //记录rpc调用时间
        long costTime = System.currentTimeMillis() - startTime;
        if (TIME_THRESHOLD < costTime) {
            log.warn("requestId:{},调用耗时:{}", request.getRequestId(), costTime);
        } else {
            log.info("requestId:{},调用耗时:{}", request.getRequestId(), costTime);
        }
    }

    private void invokeCallbacks() {
        callbackLock.lock();
        try {
            pendingCallbacks.forEach( rpcCallback -> {
                runCallback(rpcCallback);
            });
        }finally {
            callbackLock.unlock();
        }

    }

    private void runCallback(RpcCallback rpcCallback) {
        executor.submit(()->{
           if(response.getThrowable() == null){
               rpcCallback.success(this.response.getResult());
           }else {
               rpcCallback.failure(this.response.getThrowable());
           }

        });
    }


    /**
     * todo 取消逻辑
     *
     * @param mayInterruptIfRunning
     * @return
     */
    @Override
    public boolean cancel(boolean mayInterruptIfRunning) {
        throw new UnsupportedOperationException();
    }

    @Override
    public boolean isCancelled() {
        throw new UnsupportedOperationException();
    }

    @Override
    public boolean isDone() {
        return sync.isDone();
    }

    @Override
    public Object get() throws InterruptedException, ExecutionException {
        // 同步阻塞
        sync.acquire(-1);
        if (this.response != null) {
            return this.response.getResult();
        }
        return null;
    }

    /**
     * 同步阻塞方式获取服务响应时间
     *
     * @param timeout
     * @param unit
     * @return
     * @throws InterruptedException
     * @throws ExecutionException
     * @throws TimeoutException
     */
    @Override
    public Object get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
        // 快速失败同步阻塞方法
        boolean success = sync.tryAcquireNanos(-1, unit.toNanos(timeout));
        if (success) {
            if (this.response != null) {
                return this.response.getResult();
            } else {
               return null;
            }
        }
        throw new RuntimeException("timeout exception requestId: " + this.request.getRequestId()
                + ", className: " + this.request.getClassName()
                + ", methodName: " + this.request.getMethodName());
    }


    /**
     * 自定义获取锁
     */
    class Sync extends AbstractQueuedSynchronizer {

        private static final long serialVersionUID = -3109677203876043567L;

        /**
         * 已完成:1
         */
        private final int done = 1;
        /**
         * 等待
         */
        private final int pending = 0;


        @Override
        protected boolean tryAcquire(int arg) {
            return getState() == done ? true : false;
        }

        /**
         * @param arg
         * @return
         */
        @Override
        protected boolean tryRelease(int arg) {
            if (getState() == pending) {
                if (compareAndSetState(pending, done)) {
                    return true;
                }
            }
            return false;
        }

        /**
         * 判断是否完成
         *
         * @return
         */
        public boolean isDone() {
            return getState() == done;
        }
    }
}
